package basicpattern;

import java.io.IOException;

import tripleheap.InvalidSlotNumberException;
import tripleheap.InvalidTripleSizeException;
import tripleheap.THFBufMgrException;
import tripleheap.THFDiskMgrException;
import tripleheap.THFException;

import diskmgr.Stream;
import iterator.*;
import bufmgr.PageNotReadException;

import index.IndexException;
import iterator.JoinsException;
import iterator.LowMemException;
import iterator.NestedLoopException;
import iterator.PredEvalException;
import iterator.SortException;
import iterator.TupleUtils;
import iterator.TupleUtilsException;
import iterator.UnknowAttrType;
import iterator.UnknownKeyTypeException;
import global.AttrOperator;
import global.AttrType;
import global.RID;
import global.SystemDefs;
import global.TID;
import heap.Heapfile;
import heap.InvalidTupleSizeException;
import heap.InvalidTypeException;
import heap.Scan;
import heap.Tuple;

public class BpJoin extends BPIterator{
	int amt_of_mem;
	int num_left_nodes;
	BPIterator left_itr;
	int BPJoinNodePosition;
	int JoinOnSubjectorObject;
	String RightSubjectFilter;
	String RightPredicateFilter;
	String RightObjectFilter;
	Double RightConfidenceFilter;
	int [] LeftOutNodePositions;
	int OutputRightSubject;
	int OutputRightObject;
	Tuple inner_tuple;
	Tuple outer_tuple;
	AttrType[] _in2;
	AttrType[] _in1;
	short[] t2_str_sizecopy;
	short[] t1_str_sizecopy;
	Tuple JTuple;
	
	FldSpec[] perm_mat;
	int nOutFlds;
	
	private   CondExpr OutputFilter[];
	private   CondExpr RightFilter[];
	
	Heapfile  hf;
	Stream stream;
	String relationName;
	//private   Scan      inner;
	private   boolean        done,         // Is the join complete
	get_from_outer;

	public BpJoin( int amt_of_mem, int num_left_nodes, BPIterator left_itr,
			int BPJoinNodePosition, int JoinOnSubjectorObject, java.lang.String
			RightSubjectFilter, java.lang.String RightPredicateFilter, java.lang.String
			RightObjectFilter, double RightConfidenceFilter,int [] LeftOutNodePositions,
			int OutputRightSubject, int OutputRightObject) throws InvalidSlotNumberException, InvalidTripleSizeException, THFException, THFDiskMgrException, THFBufMgrException, Exception
			{
		this.amt_of_mem=amt_of_mem;
		this.num_left_nodes= num_left_nodes;
		this.left_itr = left_itr;
		this.BPJoinNodePosition = BPJoinNodePosition;
		this.JoinOnSubjectorObject = JoinOnSubjectorObject;
		this.RightSubjectFilter = RightSubjectFilter;
		this.RightPredicateFilter= RightPredicateFilter;
		this.RightObjectFilter = RightObjectFilter;
		this.RightConfidenceFilter = RightConfidenceFilter;
		this.LeftOutNodePositions = LeftOutNodePositions;
		this.OutputRightSubject = OutputRightSubject;
		this.OutputRightObject =OutputRightObject;
		//RightFilter = RightPredicateFilter;
		_in2 = new AttrType[4];
		_in2[0] = new AttrType(AttrType.attrString);
		_in2[1] = new AttrType(AttrType.attrString);
		_in2[2] = new AttrType(AttrType.attrString);
		//attrType[3] = new AttrType(AttrType.attrDouble);
		_in2[3] = new AttrType(AttrType.attrReal);
		

		t2_str_sizecopy = new short[3];
		t2_str_sizecopy[0] = 100;
		t2_str_sizecopy[1] = 100;
		t2_str_sizecopy[2] = 100;
		
		t1_str_sizecopy = new short[this.num_left_nodes];
		
		for(int i=0;i<t1_str_sizecopy.length; i++)
			t1_str_sizecopy[i] = 100;
		
		
		_in1 = new AttrType[this.num_left_nodes+1];
		int j;
		for(j=0;j<this.num_left_nodes;j++)
			_in1[j] = new AttrType(AttrType.attrString);
		
		_in1[j] = new AttrType(AttrType.attrReal);
		
		
		
		nOutFlds = this.LeftOutNodePositions.length;
		if(this.OutputRightObject == 1)
			nOutFlds++;
		if(this.OutputRightSubject == 1)
			nOutFlds++;
		nOutFlds++;
		perm_mat = new FldSpec[nOutFlds];
		int i;
		for(i=0;i<this.LeftOutNodePositions.length;i++)
			perm_mat[i] = new FldSpec(new RelSpec(RelSpec.outer), this.LeftOutNodePositions[i]);
		if(this.OutputRightSubject == 1)
			perm_mat[i++] = new FldSpec(new RelSpec(RelSpec.innerRel), 1);
		if(this.OutputRightObject == 1)
			perm_mat[i++] = new FldSpec(new RelSpec(RelSpec.innerRel), 3);
		
		perm_mat[i] = new FldSpec(new RelSpec(RelSpec.innerRel), 4);
		get_from_outer = true;
			   
		JTuple = new Tuple();
		AttrType[] Jtypes = new AttrType[nOutFlds];
		short[]    t_size;
		
		try {
			t_size = TupleUtils.setup_op_tuple(JTuple, Jtypes,
					_in1, _in1.length, _in2, _in2.length,
					t1_str_sizecopy, t2_str_sizecopy,
					perm_mat, nOutFlds);
		}catch (TupleUtilsException e){
			throw new NestedLoopException(e,"TupleUtilsException is caught by NestedLoopsJoins.java");
		}
		
		
	}

	/*    try {
    		hf = new Heapfile(relationName);
        }
        catch(Exception e) 
        {
        		throw new NestedLoopException(e, "Create new heapfile failed.");
        } */

	public Tuple get_next()
	throws IOException,
//  expr2[1].next = null;
//  expr2[1].type1 = new AttrType(AttrType.attrSymbol);
//  expr2[1].type2 = new AttrType(AttrType.attrString);
//  expr2[1].operand1.symbol = new FldSpec (new RelSpec(RelSpec.innerRel),3);
//  expr2[1].operand2.string = "red";
	JoinsException ,
	IndexException,
	InvalidTupleSizeException,
	InvalidTypeException, 
	PageNotReadException,
	TupleUtilsException, 
	PredEvalException,
	SortException,
	LowMemException,
	UnknowAttrType,
	UnknownKeyTypeException,
	Exception
	{
		if (done)
			return null;

		do
		{
			if (get_from_outer == true)
			{
				get_from_outer = false;
				if (stream != null)     // If this not the first time,
				{
					stream.close();
					stream = null;
				}
				try
				{
					stream = SystemDefs.JavabaseDB.openStream(SystemDefs.MINIBASE_DBNAME, 0, RightSubjectFilter, RightPredicateFilter, RightObjectFilter, RightConfidenceFilter, amt_of_mem);
				}
				catch(Exception e)
				{
					e.printStackTrace();
				}
				if ((outer_tuple=left_itr.get_next()) == null)
				{
					done = true;
					if (stream != null) 
					{
						stream.close();
						stream = null;
					}
					return null;
				}   
			}  // ENDS: if (get_from_outer == TRUE)
			
			outer_tuple.setHdr((short)(this.num_left_nodes+1), _in1, t1_str_sizecopy);
			RightFilter = new CondExpr[1];
			RightFilter[0] = new CondExpr();
			RightFilter[0].op   = new AttrOperator(AttrOperator.aopEQ);
			RightFilter[0].next = null;
			RightFilter[0].type1 = new AttrType(AttrType.attrString);
			RightFilter[0].type2 = new AttrType(AttrType.attrString);
			
			RightFilter[0].operand2.string = outer_tuple.getStrFld(BPJoinNodePosition);
			
			RID rid = new RID();
			while ((inner_tuple = stream.get_next(rid)) != null)
			{
				
				inner_tuple.setHdr((short)4, _in2,t2_str_sizecopy);
				
				float outerConf = outer_tuple.getFloFld(this.num_left_nodes+1);
				float innerConf = inner_tuple.getFloFld(4);
				if(outerConf < innerConf)
					perm_mat[nOutFlds-1] = new FldSpec(new RelSpec(RelSpec.outer), this.num_left_nodes+1);
				else
					perm_mat[nOutFlds -1] = new FldSpec(new RelSpec(RelSpec.innerRel), 4);
				
				if(this.JoinOnSubjectorObject == 0)
					RightFilter[0].operand1.string = inner_tuple.getStrFld(1);
				else
					RightFilter[0].operand1.string = inner_tuple.getStrFld(3);
				if (PredEval.Eval(RightFilter, outer_tuple , inner_tuple, _in1, _in2) == true)
				{ 
					//if (PredEval.Eval(OutputFilter, outer_tuple, inner_tuple, _in1, _in2) == true)
					//{
						// Apply a projection on the outer and inner tuples.
						Projection.Join(outer_tuple, _in1, 
								inner_tuple, _in2, 
								JTuple, perm_mat, nOutFlds);
						return JTuple;
					//}
				}
			}
			
			get_from_outer = true; // Loop back to top and get next outer tuple.	      
		} while (true);
		
	} 

	/**
	 * implement the abstract method close() from super class Iterator
	 *to finish cleaning up
	 *@exception IOException I/O error from lower layers
	 *@exception JoinsException join error from lower layers
	 *@exception IndexException index access error 
	 */
	public void close() throws JoinsException, IOException,IndexException 
	{
		if (!closeFlag) {
			try {
				left_itr.close();
			}catch (Exception e) {
				throw new JoinsException(e, "NestedLoopsJoin.java: error in closing iterator.");
			}
			closeFlag = true;
		}
	}
}

